package com.cdsf.common.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.IOException;
import java.net.URI;

/**
 * @Author: kennedy
 * @Date: Created in 14:28 2018/4/12
 */
public class HadoopUtil {
    private static final String HDFS_PATH = "hdfs://192.168.217.129:8020";
    private static FileSystem fileSystem;

    static {
        Configuration configuration = new Configuration();
        try {
            fileSystem = FileSystem.newInstance(new URI(HDFS_PATH), configuration, "hadoop");
        } catch (Exception e) {
            e.printStackTrace();
            fileSystem = null;
        }
    }

    /**
     * 创建文件夹
     *
     * @param path
     * @return
     */
    public static Boolean mkdirs(String path) {
        Boolean result = false;
        try {
            fileSystem.mkdirs(new Path(path));
            result = true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 上传本地文件到HDFS
     *
     * @param src
     * @param dst
     * @return
     */
    public static Boolean copyFromLocalFile(String src, String dst) {
        Boolean result = false;
        try {
            fileSystem.copyFromLocalFile(new Path(src), new Path(dst));
            result = true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 创建文件
     *
     * @param dst
     * @param context
     * @return
     */
    public static Boolean create(String dst, String context) {
        Boolean result = false;
        try {
            FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path(dst), true);
            fsDataOutputStream.write(context.getBytes());
            fsDataOutputStream.flush();
            fsDataOutputStream.close();
            result = true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 查看HDFS文件内容
     *
     * @param dst
     * @return
     */
    public static Boolean cat(String dst) {
        Boolean result = false;
        try {
            FSDataInputStream fsDataInputStream = fileSystem.open(new Path(dst));
            System.out.print("--------------------------------------------------------");
            IOUtils.copyBytes(fsDataInputStream, System.out, 1024);
            System.out.print("--------------------------------------------------------");
            fsDataInputStream.close();
            result = true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 重命名文件
     *
     * @param src
     * @param dst
     * @return
     */
    public static Boolean rename(String src, String dst) {
        Boolean result = false;
        try {
            result = fileSystem.rename(new Path(src), new Path(dst));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }
}
